package com.ym.core.rabbit;

import cn.hutool.core.date.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SerializerMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;

/**
 * 发送消息给MQ
 *
 * @author: 李涛
 * @version: 2019年09月19日 11:58
 */
@Service
public class IMessageSenderSV {

	private static final Logger LOG = LoggerFactory.getLogger(IMessageSenderSV.class);

	@Autowired
	private RabbitTemplate rabbitTemplate;

	/**
	 * 发送云信消息
	 *
	 * @param messageTask 消息内容
	 */
	public void sendMsg(NormalMessageTask messageTask) {
		LOG.info("发送[ {} ]消息到MQ", messageTask.getMessageTypeEnum().getDescribe());
		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
		rabbitTemplate.convertAndSend(RabbitMQConfig.NORMAL_EXCHANGE, RabbitMQConfig.NORMAL_QUEUE, messageTask, (message) -> {
			MessageProperties messageProperties = message.getMessageProperties();
			messageProperties.setMessageId(messageTask.getUuid());
			messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
			messageProperties.setType(messageTask.getMessageTypeEnum().getDescribe());
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
			String sendTime = DateUtil.now();
			// 发送时间
			messageProperties.setHeader("send_time", sendTime);
			return message;
		});
	}


	/**
	 * 发送延时任务给队列
	 *
	 * @param task 任务
	 */
	public void sendDelayTask(AbstractDelayedTask task) {
		LOG.info("发送延时任务 [ {}:{} ] 到MQ", task.getDescribe(), task.getDelay());
		rabbitTemplate.setMessageConverter(new SerializerMessageConverter());
		rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE, RabbitMQConfig.DELAY_QUEUE, task, (message) -> {
			MessageProperties messageProperties = message.getMessageProperties();
			long nextDelay = 0;
			if (task.getDelay() > Integer.MAX_VALUE) {
				//如果延时时间大于erlang最大数值，多次延时
				messageProperties.setDelay(Integer.MAX_VALUE);
				nextDelay = task.getDelay() - Integer.MAX_VALUE;
			} else {
				messageProperties.setDelay(task.getDelay().intValue());
			}
			// 下次延时的时间
			messageProperties.setHeader("next_delay", nextDelay);
			messageProperties.setMessageId(task.getUuid());
			messageProperties.setType(task.getDescribe());
			messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
			String sendTime = DateUtil.now();
			// 发送时间
			messageProperties.setHeader("send_time", sendTime);
			String expirationTime = DateUtil.date(new Date(System.currentTimeMillis() + task.getDelay())).toString();
			// 过期时间
			messageProperties.setHeader("expiration_time", expirationTime);
			// 任务的入参
			messageProperties.setHeader("params", task.getParams().toString());
			return message;
		});
	}

	/**
	 * 多次延时，再次发送任务
	 *
	 * @param task      任务
	 * @param nextDelay 下次延时时间
	 */
	public void sendAgain(Message task, final long nextDelay) {
		rabbitTemplate.setMessageConverter(new SerializerMessageConverter());
		rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE, RabbitMQConfig.DELAY_QUEUE, task, (message) -> {
			MessageProperties messageProperties = message.getMessageProperties();
			long nextDelayNew = 0;
			if (nextDelay > Integer.MAX_VALUE) {
				//如果延时时间大于erlang最大数值，多次延时
				messageProperties.setDelay(Integer.MAX_VALUE);
				nextDelayNew = nextDelay - Integer.MAX_VALUE;
			} else {
				messageProperties.setDelay((int) nextDelay);
			}
			// 下次延时的时间
			messageProperties.setHeader("next_delay", nextDelayNew);
			return message;
		});
	}
}
